package HBase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;
import java.util.*;

/**
 * @Author lixinlei
 * @Date 2023/3/30 8:51
 */
public class Demo1_NamespaceOperator {

    static Connection conn = null;
    static Admin admin = null;

    static{

        try {
            // 在 Hadoop 的 Configuration 基础上 添加了 hbase-default和hbase-site两个配置文件的Configuration对象
            Configuration conf = HBaseConfiguration.create();

            //把连接zookeeper的相关信息设置进去
            conf.set(Constant.ZK_QUORUM,Constant.ZK_NODES);
            conf.set(Constant.ZK_CLIENTPORT,Constant.ZK_PORT);

            //携带zookeeper信息创建连接对象
            conn = ConnectionFactory.createConnection(conf);

            //通过连接创建namespace的操作对象
            admin = conn.getAdmin();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 创建一个没有属性的namespace
     * @param nsName
     */
    public static void createNamespace(String nsName){
        createNamespace(nsName,null);
    }

    /**
     * 创建一个有不定长个属性的namespace
     * @param nsName
     * @param propertiesMap
     */
    public static void createNamespace(String nsName, Map<String,String> propertiesMap){
        try {

            //判断当前的namespace是否已经存在
            NamespaceDescriptor nsd_old = admin.getNamespaceDescriptor(nsName);

            System.out.println("当前"+nsName+" namespace已经存在，无法添加");

        }catch (NamespaceNotFoundException e){
            try {
                //如果namespace不存在，可以创建此namespace
                NamespaceDescriptor nsd = NamespaceDescriptor.create(nsName).build();

                //如果有属性的话，添加属性值，没有就不添加
                if(null!=propertiesMap && propertiesMap.size()>0){
                    Set<Map.Entry<String, String>> entries = propertiesMap.entrySet();
                    for (Map.Entry<String, String> entry : entries) {
                        nsd.setConfiguration(entry.getKey(),entry.getValue());
                    }
                }
                admin.createNamespace(nsd);
            }catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            Tools.closed(admin,conn);
        }

    }


    public static void alterNamespace(String nsName,Object... properties){


        try {
            //判断当前的namespace是否存在
            NamespaceDescriptor nsd_old = admin.getNamespaceDescriptor(nsName);
            //如果当前namespace存在的话，获取所有的属性内容
            Map<String, String> confAll = nsd_old.getConfiguration();

            //添加修改删除所有的操作一起执行
            for (Object property : properties) {
                //如果参数是Map类型，做添加或修改属性的操作
                if(property instanceof Map){
                    //确定类型后，强转为Map类型
                    Map<String,String> propMap = (Map<String, String>) property;
                    Set<Map.Entry<String, String>> entries = propMap.entrySet();
                    //遍历所有需要添加或修改的属性，有nsd_old执行
                    for (Map.Entry<String, String> entry : entries) {
                        nsd_old.setConfiguration(entry.getKey(),entry.getValue());
                    }
                } else if (property instanceof List) {
                    //如果参数是List类型，做删除属性的操作

                    //确定类型后，强转为List类型
                    List<String> propList = (List<String>) property;
                    for (String prop : propList) {
                        //判断当前需要删除的key是否存在，存在才可以删除
                        if(confAll.containsKey(prop)){
                            nsd_old.removeConfiguration(prop);
                        }else{
                            System.out.println("当前"+prop+"属性不存在，无法删除");
                        }
                    }
                }
            }

            //最后执行所有的关于属性增删改的操作
            admin.modifyNamespace(nsd_old);

        } catch (NamespaceNotFoundException e) {
            System.out.println("当前"+nsName+" namespace不存在，不能修改");
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }


    /**
     * 获取某个namespace下的所有属性
     * @param nsName
     */
    public static void getNamespaceProperties(String nsName){
        getNamespaceProperties(nsName,null);
    }

    /**
     * 获取某个namespace下的部分属性
     * @param nsName
     * @param propKeys
     */
    public static void getNamespaceProperties(String nsName,List<String> propKeys){

        try {
            //判断当前的namespace是否存在
            NamespaceDescriptor nsd_old = admin.getNamespaceDescriptor(nsName);

            //因为需要判断想要获取的key是否存在，先把所有的key获取，方便后面的判断
            Map<String, String> confAll = nsd_old.getConfiguration();

            //如果propKeys参数不为空，就是获取指定属性值
            if (propKeys!=null && propKeys.size()>0) {
                //遍历获取所有需要获取的值
                for (String propKey : propKeys) {
                    //如果目标属性不存在也无法获取
                    if(confAll.containsKey(propKey)){
                        System.out.println(propKey+":"+nsd_old.getConfigurationValue(propKey));
                    }else{
                        System.out.println("当前"+propKey+"属性不存在，无法获取");
                    }
                }
            }else{
                //如果propKeys为空获取所有属性值
                if(confAll!=null && confAll.size()>0){
                    Set<Map.Entry<String, String>> entries = confAll.entrySet();
                    for (Map.Entry<String, String> entry : entries) {
                        System.out.println(entry.getKey()+":"+entry.getValue());
                    }
                }else{
                    System.out.println("当前"+nsName+"namespace没有属性");
                }
            }

        } catch (NamespaceNotFoundException e) {
            System.out.println("当前"+nsName+" namespace不存在，无法查询");
        } catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }


    /**
     * 获取所有namespace下的所有表
     */
    public static void getTablesByNamespace(){
        getTablesByNamespace(null);
    }

    /**
     * 获取执行namespace下的所有表
     * @param nsName
     */
    public static void getTablesByNamespace(String nsName){

        //初始化所有表的数组
        TableName[] tableNames = null;

        try{
            //如果指定了namespace
            if(nsName!=null && !nsName.equals("")){
                //先判断当前namespace是否存在
                if(Tools.isNamespaceExists(admin,nsName)){
                    //如果存在，获取执行namespace下的所有表
                    tableNames = admin.listTableNamesByNamespace(nsName);
                }
            }else{
                //如果没指定namespace，获取所有namespace下的所有表
                tableNames = admin.listTableNames();
            }


            //不管哪种情况，遍历所有的表
            for (TableName tableName : tableNames) {
                System.out.println("tableName:"+tableName.getNameAsString());
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }

    /**
     * 获取当前hbase下的所有namespace
     */
    public static void getAllNamespace(){
        try{
            //获取所有的namespaceDescriptor
            NamespaceDescriptor[] nsds = admin.listNamespaceDescriptors();

            //如果有namespace，遍历输出
            if(nsds!=null && nsds.length>0){
                for (NamespaceDescriptor nsd : nsds) {
                    String nsName = nsd.getName();
                    System.out.println("nsName:"+nsName);
                }
            }else{
                System.out.println("hbase中没有namespace");
            }


        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }

    }

    public static void dropNamespace(String nsName,boolean r){

    }

    public static void dropNamespace(String nsName){

        try{
            //删除的第一个条件就是namespace必须存在
            if(Tools.isNamespaceExists(admin,nsName)){
                //获取当前namespace下的所有表，用来判断是否非空
                TableName[] tableNames = admin.listTableNamesByNamespace(nsName);
                //如果是空namespace，可以删除
                if(tableNames==null || tableNames.length==0){
                    admin.deleteNamespace(nsName);
                }else{
                    System.out.println(nsName+" namespace非空，无法删除");
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }


    }

    public static void dropNamespace(List<String> nsNameList){

        //此处没有必须需要处理的异常，可以加try...catch,用finally关闭连接。也可以不加try直接在代码最后关闭连接
        try{
            //如果列表有值，遍历删除指定的namespace
            if(null!=nsNameList && nsNameList.size()>0){
                for (String nsName : nsNameList) {
                    dropNamespace(nsName);
                }
            }else{
                System.out.println("列表未指定，无法删除");
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            Tools.closed(admin,conn);
        }
    }

    public static void main(String[] args) {

        //1.创建namespace
//        createNamespace("test1",propMap);

        //2.修改namespace的属性
//        Map<String,String> addOrModifyMap = new HashMap<>();
//        addOrModifyMap.put("NAME","ztq");
//        addOrModifyMap.put("AGE","10000");
//        addOrModifyMap.put("gender","unknown");
//
//        List<String> deleteList = new ArrayList<>();
//        deleteList.add("ID");
//        deleteList.add("aaaa");
//
//        alterNamespace("test1",addOrModifyMap,deleteList);

        //3.获取属性
//        List<String> propList = new ArrayList<>();
//        propList.add("gender");
//        propList.add("address");
//
//        getNamespaceProperties("test1",propList);

        //4.获取表
//        getTablesByNamespace();
//        getTablesByNamespace("test1");

        //5.获取hbase中的所有namespace
//        getAllNamespace();

        //6.删除namespace
//        dropNamespace("test");

//        List<String> deleteNSList = new ArrayList<>();
//        deleteNSList.add("r");
//        deleteNSList.add("w");
//        deleteNSList.add("t");
//        deleteNSList.add("test1");
//        dropNamespace(deleteNSList);


    }

}
